package pers.wk.consumer;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;
import pers.wk.exceptions.BizException;
import pers.wk.rabbitmq.Constants;
import pers.wk.redis.RedisService;

import java.util.Map;

/**
 * 注解方式配置
 * 发送站内信消费者
 * 消息ack的方式为MANUAL
 * 用户需要手动发送ack或者nack
 */
@Slf4j
@Component
@Qualifier("znxConsumer")
public class ZnxConsumer {

    @Autowired
    private RedisService redisService;

    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = "${msg.znx.queue}", durable = "true",
                            arguments = {@Argument(name = "", value = ""), @Argument(name = "", value = "")}),
                    exchange = @Exchange(value = "${msg.znx.exchange}", type = ExchangeTypes.DIRECT),
                    key = "${msg.znx.binding-key}"
            )
    )
    public void sendZnx(Message message, Channel channel) throws Exception {
        MessageProperties messageProperties = message.getMessageProperties();
        Map<String, Object> headers = messageProperties.getHeaders();
        String msgId = (String) headers.get(Constants.MSG_ID);
        String msg = new String(message.getBody());
        log.info(String.format("znxQueue 收到的消息为: msgId: {%s} {%s}", msgId, msg));
        try {

            // 这里写各种业务逻辑
            //1.验证消息是否被重复消费
            if (!redisService.setIfAbsent(msgId, msg)) {
                throw new BizException("消息重复消费");
            }
            //2.模拟发送站内信
            log.info("发送站内信通知：{}", msg);
            //若在此处不做应答，在消费者重启后，会继续消费该消息，因为做了消费幂等性处理，所以重复消费，会被消息直接丢弃
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            e.printStackTrace();
            //消费失败，消息丢弃
            if (e instanceof BizException) {
                log.info("消息重复消费：【msgId: {}】,【msg: {}】", msgId, msg);
            }
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        }
    }

    /**
     * 在MQ服务中，提前定义好队列交换机绑定关系，可不在代码中定义
     *
     * @param message
     * @param channel
     */
    //@RabbitListener(queues = {"${msg.znx.queue}"})
    public void sendZnx2(Message message, Channel channel) {
        //同上
    }
}
